package com.corn.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.alibaba.otter.canal.protocol.CanalEntry.EntryType.ROWDATA;
import static com.alibaba.otter.canal.protocol.CanalEntry.EventType.*;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2022/7/15 11:12
 */
@Slf4j
public class CanalTest {

    public static void main(String[] args) throws InterruptedException {
        // 创建链接
        /**
         * address: MySQL 链接
         * destination: 对应canal中的dest配置
         * username: canal客户端链接canal服务端的账号密码不是MySQL的密码!!! 默认为空
         * password:
         */
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.209.100",
                11111), "example", "", "");
        // 客户端链接canal服务器
        connector.connect();
        // 订阅监听的数据库名称
        connector.subscribe("canal_test.*");
        while (true) {
            // 每次请求100条变更
            Message message = connector.get(100);
            // 如果没有获取到信息 休眠3sec
            // 变更的记录封装在Message的entry中
            List<CanalEntry.Entry> entries = message.getEntries();
            if (entries.size() == 0) {
                log.info("没有获取到数据");
                Thread.sleep(3000);
            }
            entries.forEach(entry -> {
                // 如果是数据变更
                /**
                 * ROWDATA 数据变更
                 * TRANSACTIONBEGIN 事务开启
                 * TRANSACTIONEND 事务关闭
                 * 一般我们关心的是ROWDATA
                 */
                if (entry.getEntryType() == ROWDATA) {
                    // entry中的storeValue是对应binlog日志中的序列化之后的数据 使用需要反序列化
                    ByteString storeValue = entry.getStoreValue();
                    try {
                        // 调用RowChange.parseFrom 对数据进行反序列化
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                        printRowChange(rowChange);
                    } catch (InvalidProtocolBufferException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

    }

    public static void printRowChange(CanalEntry.RowChange rowChange) {
        // rowChange中的事件包含了INSERT DELETE UPDATE ALTER等等..一般关心增删改
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        if (rowChange.getEventType() == INSERT) {
            log.info("=================== 插入操作 ===================");
        } else if (rowChange.getEventType() == DELETE) {
            log.info("=================== 删除操作 ===================");
        } else if (rowChange.getEventType() == UPDATE) {
            log.info("=================== 更新操作 ===================");
        }
        rowDatasList.forEach(data -> {
            // 变更前的数据集合 RowData.getBeforeColumnsList 封装出来封装成column 中的name是MySQL的字段名 value就是对应的数据
            List<CanalEntry.Column> beforeColumnsList = data.getBeforeColumnsList();
            log.info("=================== before ===================");
            Map<String, String> beforeData = beforeColumnsList.stream().collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
            log.info("变更前的数据:{}", beforeData.toString());
            log.info("=================== before end ===================");
            // 变更后的数据集合
            List<CanalEntry.Column> afterColumnsList = data.getAfterColumnsList();
            log.info("=================== after ===================");
            Map<String, String> afterData = afterColumnsList.stream().collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
            log.info("变更后的数据:{}", afterData.toString());
            log.info("=================== after end ===================");
        });
    }

}
